import json
import random
import time
from datetime import datetime, timedelta
import threading
from queue import Queue
from kafka import KafkaProducer

# bootstrap_servers = "106.120.201.126:14576"
bootstrap_servers = "192.168.1.101:59092"
b5_topic = "b5_data_wanji"
status_topic = "device_status"


# 定义设备列表
up_list = ["101", "102", "103", "104", "105"]
down_list = ["201", "202", "203", "204", "205"]

gantry_info = {
    "up": {
        "10001": {
            "up_gantry_id": ["101"],
            "down_gantry_id": ["102"],
            "direction": 0,
            "lane_num": 2,
            "dist": 0.885,
            "car_max_speed": 80,
            "is_fork": 0
        },
        "10002": {
            "up_gantry_id": ["102"],
            "down_gantry_id": ["103"],
            "direction": 0,
            "lane_num": 2,
            "dist": 0.615,
            "car_max_speed": 80,
            "is_fork": 1
        },
        "10003": {
            "up_gantry_id": ["103"],
            "down_gantry_id": ["104"],
            "direction": 0,
            "lane_num": 2,
            "dist": 0.865,
            "car_max_speed": 80,
            "is_fork": 1
        },
        "10004": {
            "up_gantry_id": ["104"],
            "down_gantry_id": ["105"],
            "direction": 0,
            "lane_num": 2,
            "dist": 0.93,
            "car_max_speed": 80,
            "is_fork": 0
        }
    },
    "down": {
        "20001": {
            "up_gantry_id": ["201"],
            "down_gantry_id": ["202"],
            "direction": 0,
            "lane_num": 2,
            "dist": 0.865,
            "car_max_speed": 80,
            "is_fork": 1
        },
        "20002": {
            "up_gantry_id": ["202"],
            "down_gantry_id": ["203"],
            "direction": 0,
            "lane_num": 2,
            "dist": 0.615,
            "car_max_speed": 80,
            "is_fork": 1
        },
        "20003": {
            "up_gantry_id": ["203"],
            "down_gantry_id": ["204"],
            "direction": 0,
            "lane_num": 2,
            "dist": 0.12,
            "car_max_speed": 80,
            "is_fork": 1
        },
        "20004": {
            "up_gantry_id": ["204"],
            "down_gantry_id": ["205"],
            "direction": 0,
            "lane_num": 2,
            "dist": 0.765,
            "car_max_speed": 80,
            "is_fork": 0
        }
    }
}

device_info = {
    "up": {
        "101": {
            "cameraIP": "10.144.26.3",
            "longitude": 105.557003580743,
            "latitude": 27.9728598243572,
            "pileNumber": "K1757+865"
        },
        "102": {
            "cameraIP": "10.144.24.174",
            "longitude": 105.54952927,
            "latitude": 27.97709098,
            "pileNumber": "K1758+750"
        },
        "103": {
            "cameraIP": "10.144.24.133",
            "longitude": 105.54565288,
            "latitude": 27.98142537,
            "pileNumber": "K1759+365"
        },
        "104": {
            "cameraIP": "10.144.24.94",
            "longitude": 105.54460953,
            "latitude": 27.98888572,
            "pileNumber": "K1760+230"
        },
        "105": {
            "cameraIP": "10.144.24.81",
            "longitude": 105.55185591,
            "latitude": 27.99401906,
            "pileNumber": "K1761+160"
        }
    },
    "down": {
        "201": {
            "cameraIP": "10.144.24.97",
            "longitude": 105.54415942,
            "latitude": 27.98748340,
            "pileNumber": "K1760+230"
        },
        "202": {
            "cameraIP": "10.144.24.136",
            "longitude": 105.54555833,
            "latitude": 27.98138341,
            "pileNumber": "K1759+365"
        },
        "203": {
            "cameraIP": "10.144.24.175",
            "longitude": 105.54948796,
            "latitude": 27.97696681,
            "pileNumber": "K1758+750"
        },
        "204": {
            "cameraIP": "10.144.24.187",
            "longitude": 105.55038027,
            "latitude": 27.97621773,
            "pileNumber": "K1758+630"
        },
        "205": {
            "cameraIP": "10.144.26.6",
            "longitude": 105.55698531,
            "latitude": 27.97274828,
            "pileNumber": "K1757+865"
        }
    },
}

# 共享事件ID记录
shared_event_ids = {}
event_lock = threading.Lock()
event_dict_str = {1: "畅通", 2: "轻度拥堵", 3: "中度拥堵", 4: "严重拥堵", 5: "阻断"}


# 全局事件ID生成器
class EventIdGenerator:
    def __init__(self):
        self.lock = threading.Lock()
        self.current_date = datetime.now().strftime('%Y%m%d')
        self.current_index = 1

    def get_next_id(self):
        with self.lock:
            # 检查是否需要重置序号
            current_date = datetime.now().strftime('%Y%m%d')
            if current_date != self.current_date:
                self.current_date = current_date
                self.current_index = 1

            # 生成格式化的序号（3位数字，不足补零）
            index_str = f"{self.current_index:03d}"
            self.current_index += 1

            return f"{self.current_date}{index_str}"


# 初始化事件ID生成器
event_id_generator = EventIdGenerator()


def threading_main(event_queue, status_queue, flag, i):
    global shared_event_ids
    last_event_time = int(time.time())
    last_status_time = int(time.time())
    try:
        # 获取龙门架信息
        if flag == "up":
            key = str(10000 + i)
            if key not in gantry_info["up"]:
                print(f"警告: 上行方向中未找到龙门架ID {key}")
                return
            gantry = gantry_info["up"][key]
        else:
            key = str(20000 + i)
            if key not in gantry_info["down"]:
                print(f"警告: 下行方向中未找到龙门架ID {key}")
                return
            gantry = gantry_info["down"][key]

        # 获取上下行龙门架ID
        up_gantry_id = gantry["up_gantry_id"][0]
        down_gantry_id = gantry["down_gantry_id"][0]

        while True:
            time.sleep(0.1)
            if int(time.time()) - last_status_time >= 10:
                time.sleep(1)
                status_data = {
                    "deviceId": down_gantry_id,
                    "heartbeat": 1,
                    "clock": 1,
                    "delay": 1,
                    "precision": 90
                }
                status_queue.put(status_data)
                if i == 1:
                    status_data = {
                        "deviceId": up_gantry_id,
                        "heartbeat": 1,
                        "clock": 1,
                        "delay": 1,
                        "precision": 90
                    }
                status_queue.put(status_data)
                last_status_time = int(time.time())

            if int(time.time()) - last_event_time >= 30:
                time.sleep(1)
                ff = random.randint(1, 3)

                if ff == 1:
                    # 检查该区间是否已有事件ID
                    event_key = f"{up_gantry_id}_{down_gantry_id}"

                    with event_lock:
                        if event_key in shared_event_ids and datetime.now() - shared_event_ids[event_key]["timestamp"] < timedelta(minutes=1):
                            event_id = shared_event_ids[event_key]["event_id"]
                            congestion_level = random.randint(2, 5)
                            is_new_event = False
                        else:
                            # 生成新的事件ID和拥堵等级
                            event_id = event_id_generator.get_next_id()
                            event_id = int(event_id)
                            congestion_level = random.randint(2, 5)
                            shared_event_ids[event_key] = {
                                "event_id": event_id,
                                "congestion_level": congestion_level,
                                "timestamp": datetime.now()
                            }
                            is_new_event = True

                        # 获取设备位置信息
                        up_device = device_info[flag].get(up_gantry_id, {})
                        down_device = device_info[flag].get(down_gantry_id, {})

                        if shared_event_ids[event_key]["congestion_level"] != 5:
                            isBreakdown = 0
                            secLevel = shared_event_ids[event_key]["congestion_level"]
                        else:
                            isBreakdown = 1
                            secLevel = 4
                        # 生成拥堵事件数据
                        event_data = {
                            "timestamp": int(datetime.now().timestamp()),
                            "direction": 0 if flag == "up" else 1,
                            "sNum": up_device["pileNumber"],
                            "eNum": down_device["pileNumber"],
                            "sLon": up_device["longitude"],
                            "sLat": up_device["latitude"],
                            "eLon": down_device["longitude"],
                            "eLat": down_device["latitude"],
                            "secLevel": secLevel,
                            "isBreakdown": isBreakdown,
                            "eventId": event_id,
                            "eventInfo": event_dict_str[congestion_level]
                        }

                    # 将事件放入队列
                    event_queue.put(event_data)

                last_event_time = int(time.time())

    except Exception as e:
        print(f"线程 [{flag}-{i}] 发生错误: {str(e)}")


def main():
    print("开始生成并上报模拟的拥堵数据:")
    event_queue = Queue()
    status_queue = Queue()
    producer = KafkaProducer(
        bootstrap_servers=bootstrap_servers,
        value_serializer=lambda x: json.dumps(x).encode('utf-8')
    )

    # 创建并启动线程
    threads = []
    for i in range(1, 5):  # 龙门架ID从10001/20001开始
        # 上行线程
        up_thread = threading.Thread(target=threading_main,args=(event_queue, status_queue, "up", i))
        threads.append(up_thread)
        up_thread.start()
        time.sleep(0.5)  # 避免时间戳完全相同

        # 下行线程
        down_thread = threading.Thread(target=threading_main, args=(event_queue, status_queue, "down", i))
        threads.append(down_thread)
        down_thread.start()
        time.sleep(0.5)  # 避免时间戳完全相同

    # device_list = ["101", "102", "103", "104", "105", "201", "202", "203", "204", "205"]
    # status_results = {}
    # for k in device_list:
    #     status_results[k] = ({"deviceId": k, "heartbeat": 0, "clock": 0, "delay": 0, "precision": 100})
    l_time = int(time.time())
    while True:
        time.sleep(1)
        # 处理事件队列
        # print("\n处理事件队列:")
        # event_results = []
        while not event_queue.empty():
            event = event_queue.get(block=False)
            producer.send(b5_topic, value=event)
            print(f"发送事件: {event}")

        # # 处理状态队列
        # while not status_queue.empty():
        #     s = status_queue.get()
        #     status_results[s["deviceId"]] = s

        if int(time.time()) - l_time >= 10:
            l_time = int(time.time())
            # status_results_k = []
            # for k in device_list:
            #     status_results_k.append(status_results[k])

            status_results = [{'deviceId': '101', 'heartbeat': 0, 'clock': 0, 'delay': 0, 'precision': 100}, {'deviceId': '102', 'heartbeat': 0, 'clock': 0, 'delay': 0, 'precision': 95},
             {'deviceId': '103', 'heartbeat': 0, 'clock': 0, 'delay': 0, 'precision': 80}, {'deviceId': '104', 'heartbeat': 0, 'clock': 0, 'delay': 0, 'precision': 50},
             {'deviceId': '105', 'heartbeat': 1, 'clock': 0, 'delay': 0, 'precision': 100}, {'deviceId': '201', 'heartbeat': 0, 'clock': 0, 'delay': 0, 'precision': 100},
             {'deviceId': '202', 'heartbeat': 1, 'clock': 0, 'delay': 0, 'precision': 100}, {'deviceId': '203', 'heartbeat': 0, 'clock': 1, 'delay': 1, 'precision': 90},
             {'deviceId': '204', 'heartbeat': 0, 'clock': 1, 'delay': 1, 'precision': 80}, {'deviceId': '205', 'heartbeat': 0, 'clock': 1, 'delay': 1, 'precision': 65}]
            status = {
                "timestamp": int(time.time()),
                "deviceId": status_results
            }
            producer.send(status_topic, value=status)
            print(f"发送状态: {status}")


if __name__ == "__main__":
    main()